package cn.enjoy.rabbitmq.config;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

/**
 * 消费端限流，监听器--消费者
 *
 * @author K
 */
@Slf4j
@Component
public class ManualAckListener implements ChannelAwareMessageListener {

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //因为传递消息的时候用的map传递,所以将Map从Message内取出需要做些处理
            String msg = message.toString();
            //可以点进Message里面看源码,单引号直接的数据就是我们的map消息数据
            String[] msgArray = msg.split("'");
            Map<String, String> msgMap = mapStringToMap(msgArray[1].trim());
            String messageId = msgMap.get("messageId");
            String messageData = msgMap.get("messageData");
            String createTime = msgMap.get("createTime");

            // 也可以直接传json字符串，然后再解析一下
            // String json = new String(message.getBody(), StandardCharsets.UTF_8);
            // MerchantDTO merchantDTO = JsonUtil.parse(message.getBody(), MerchantDTO.class);

            if ("manualAck.queue".equals(message.getMessageProperties().getConsumerQueue())) {
                System.out.println("消费的消息来自的队列名为：" + message.getMessageProperties().getConsumerQueue());
                System.out.println("消息成功消费  messageId:" + messageId + "  messageData:" + messageData + "  createTime:" + createTime);
                System.out.println("执行manualAck.queue中的消息的业务处理流程......");
            }

            if ("fanout.A".equals(message.getMessageProperties().getConsumerQueue())) {
                System.out.println("消费的消息来自的队列名为：" + message.getMessageProperties().getConsumerQueue());
                System.out.println("消息成功消费到  messageId:" + messageId + "  messageData:" + messageData + "  createTime:" + createTime);
                System.out.println("执行fanout.A中的消息的业务处理流程......");

            }

            //参数二multiple批处理 true：可以一次性确认 delivery_tag 小于等于传入值的所有消息；false：只消费当前的deliveryTag消息
            channel.basicAck(deliveryTag, true);
//            Thread.sleep(3000);
        } catch (Exception e) {
            // 参数二requeue true：会被放回队列，重新排队，然后又会被消费；false：丢弃成为死信
            // 当消费失败，也可以记录到通用日志
            channel.basicReject(deliveryTag, false);
            log.error(e.toString());
        }
    }

    /**
     * {key=value,key=value,key=value} 格式转换成map
     */
    private Map<String, String> mapStringToMap(String str) {
        str = str.substring(1, str.length() - 1);
        String[] strs = str.split(",", 3);
        Map<String, String> map = new HashMap<String, String>();
        for (String string : strs) {
            String key = string.split("=")[0].trim();
            String value = string.split("=")[1];
            map.put(key, value);
        }
        return map;
    }
}